package org.apache.hadoop.hdfs.shortcircuit;

import java.io.BufferedOutputStream;
import java.io.Closeable;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.MappedByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang.mutable.MutableBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.ExtendedBlockId;
import org.apache.hadoop.hdfs.net.DomainPeer;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
import org.apache.hadoop.hdfs.protocolPB.PBHelper;
import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.RetriableException;
import org.apache.hadoop.net.unix.DomainSocket;
import org.apache.hadoop.net.unix.DomainSocketWatcher;
import org.apache.hadoop.security.token.SecretManager.InvalidToken;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.Waitable;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * ShortCircuitCache是DFSClient短路读取操作中最重要的类，负责对DFSClient的所有ShortCircuitReplica对象进行缓存以及生命周期管理等操作
 *
 * ![](https://markdown-sjc.oss-cn-beijing.aliyuncs.com/img/20210618174639.png)
 *
 * ShortCircuitReplica在ShortCircuitCache中会处于如下几种状态。
 *     normal：ShortCircuitReplica最正常状态，可以随时通过fetch()方法从缓存中取出。
 *     evictable：可移除状态，当DFSClient中只有ShortCircuitCache类引用了该副本，没有其他类引用时，副本处于可移除状态。ShortCircuitCache会定期清理缓存中的可移除副本。
 *     purged：删除状态，副本已经从缓存中删除了，DFSClient不能通过缓存获取该副本。处于删除状态的副本不一定被关闭了，有可能还有引用这个副本的操作，当引用数等于0时，就可以将副本关闭了。
 *     closed：关闭状态，副本对应的Slot已经从共享内存中释放了。
 *     stale：过期状态，有可能是副本对应的Slot无效，也有可能是副本对应的输入流异常，还有可能是副本在缓存中存在的时间超过有效时间。
 *
 * 一下可能发生状态转移的情况。
 *     - 起始 -> normal：用户调用fetchAndCreat()方法创建一个副本时，这时副本处于缓存中，则副本为正常状态。
 *     - normal -> stale：
 *         - fetch()副本时发现副本对应的Slot无效，或者是副本在缓存中存在的时间超过有效时间，则将副本状态设置为stale状态。
 *         - unref()操作时发现副本对应的输入流异常，则将副本状态设置为stale状态。
 *     - stale -> purged：一旦发现stale状态的副本，就马上调用purge()方法将副本从缓存中删除，将副本状态设置为purged状态。
 *     - normal -> evictable：unref()操作时，发现refCount==1，也就是只有ShortCircuitCache引用了当前副本，这时就将副本放入evictable队列中，将副本状态设置为evictable状态。
 *     - evictable -> normal：ref()操作时，refCount增加，则将副本移出evictable队列，将副本状态恢复为正常状态。
 *     - evictable -> purged：
 *         - CacheCleaner发现evictable队列中有超时副本，则将超时副本删除。
 *         - trimEnviacationMap()：当我们向evictable队列中添加元素时，会触发trimEnviacationMap()方法判断是否超过缓存数量。如果超过，则从缓存队列中直接删除副本。
 *     - purged -> closed：当refCount ==0 时，也就是完全没有对象引用这个副本时，调用副本的close()方法。
 *
 *
 */
@InterfaceAudience.Private
public class ShortCircuitCache implements Closeable {
  public static final Log LOG = LogFactory.getLog(ShortCircuitCache.class);

  /**
   *
   * 通过purge() 和 unref()方法知道，
   * 当缓存中的副本没有被对象引用时，就会被放入eviCtable队列中。
   * 由于有些副本执行了零拷贝读操作，也就是副本的数据块文件被映射到了内存中，这种已经进行了mmap的副本，我们会将它放入evictableMmapped队列中保存，而没有进行mmap操作的副本则放入另一个队列evictable中。
   * 当需要从队列中删除元素时，优先从evictable队列中删除，因为mmap操作更加耗时。
   *
   * 向evictable以及evictableMmapped中添加元素，已经介绍过了， 当副本的refCount==1时，就会被添加到evictable队列中。
   *
   * 那什么时候删除呢？
   * 当ShortCircuitCache创建了一个副本时，就会尝试启动CacheCleaner线程。
   */
  private class CacheCleaner implements Runnable, Closeable {
    private ScheduledFuture<?> future;

    /**
     * CacheCleaner线程会尝试调用demoteOldEvictableMmaped()方法将evictableMmapped中的超时副本（超过maxEvictableMmapedLifespanMs时间）放入evictable队列中，
     * 然后将evictable队列中的超时副本（超过maxNonMmappedEvictableLifespanMs时间）从缓存中删除。
     */
    @Override
    public void run() {
      ShortCircuitCache.this.lock.lock();
      try {
        if (ShortCircuitCache.this.closed) {
          return;
        }
        long curMs = Time.monotonicNow();
        // 将 evictableMmapped队列中的元素放入 evictable队列中
        int numDemoted = demoteOldEvictableMmaped(curMs);
        int numPurged = 0;
        Long evictionTimeNs = Long.valueOf(0);
        while (true) {
          // evictable 是 treeMap,按序拿出第一个副本
          Entry<Long, ShortCircuitReplica> entry =  evictable.ceilingEntry(evictionTimeNs);
          if (entry == null) {
            break;
          }
          evictionTimeNs = entry.getKey();
          long evictionTimeMs =  TimeUnit.MILLISECONDS.convert(evictionTimeNs, TimeUnit.NANOSECONDS);
          // 大于 maxNonMmappedEvictableLifespanMs 时间的，则直接删除
          if (evictionTimeMs + maxNonMmappedEvictableLifespanMs >= curMs) {
            break;
          }
          ShortCircuitReplica replica = entry.getValue();
          // 调用 purge()方法删除这个副本
          purge(replica);
          numPurged++;
        }
      } finally {
        ShortCircuitCache.this.lock.unlock();
      }
    }

    @Override
    public void close() throws IOException {
      if (future != null) {
        future.cancel(false);
      }
    }

    public void setFuture(ScheduledFuture<?> future) {
      this.future = future;
    }

    /**
     * Get the rate at which this cleaner thread should be scheduled.
     *
     * We do this by taking the minimum expiration time and dividing by 4.
     *
     * @return the rate in milliseconds at which this thread should be scheduled.
     */
    public long getRateInMs() {
      long minLifespanMs = Math.min(maxNonMmappedEvictableLifespanMs, maxEvictableMmapedLifespanMs);
      long sampleTimeMs = minLifespanMs / 4;
      return (sampleTimeMs < 1) ? 1 : sampleTimeMs;
    }
  }

  /**
   * A task which asks the DataNode to release a short-circuit shared memory
   * slot.  If successful, this will tell the DataNode to stop monitoring
   * changes to the mlock status of the replica associated with the slot.
   * It will also allow us (the client) to re-use this slot for another
   * replica.  If we can't communicate with the DataNode for some reason,
   * we tear down the shared memory segment to avoid being in an inconsistent
   * state.
   */
  private class SlotReleaser implements Runnable {
    /**
     * The slot that we need to release.
     */
    private final Slot slot;

    SlotReleaser(Slot slot) {
      this.slot = slot;
    }

    @Override
    public void run() {
      final DfsClientShm shm = (DfsClientShm)slot.getShm();
      final DomainSocket shmSock = shm.getPeer().getDomainSocket();
      DomainSocket sock = null;
      DataOutputStream out = null;
      final String path = shmSock.getPath();
      boolean success = false;
      try {
        sock = DomainSocket.connect(path);
        out = new DataOutputStream( new BufferedOutputStream(sock.getOutputStream()));
        new Sender(out).releaseShortCircuitFds(slot.getSlotId());
        DataInputStream in = new DataInputStream(sock.getInputStream());
        ReleaseShortCircuitAccessResponseProto resp = ReleaseShortCircuitAccessResponseProto.parseFrom( PBHelper.vintPrefixed(in));
        if (resp.getStatus() != Status.SUCCESS) {
          String error = resp.hasError() ? resp.getError() : "(unknown)";
          throw new IOException(resp.getStatus().toString() + ": " + error);
        }
        success = true;
      } catch (IOException e) {
        LOG.error(ShortCircuitCache.this + ": failed to release " + "short-circuit shared memory slot " + slot + " by sending " +
            "ReleaseShortCircuitAccessRequestProto to " + path + ".  Closing shared memory segment.", e);
      } finally {
        if (success) {
          shmManager.freeSlot(slot);
        } else {
          shm.getEndpointShmManager().shutdown(shm);
        }
        IOUtils.cleanup(LOG, sock, out);
      }
    }
  }

  public interface ShortCircuitReplicaCreator {
    /**
     * Attempt to create a ShortCircuitReplica object.
     * This callback will be made without holding any locks.
     * @return a non-null ShortCircuitReplicaInfo object.
     */
    ShortCircuitReplicaInfo createShortCircuitReplicaInfo();
  }

  /**
   * Lock protecting the cache.
   */
  private final ReentrantLock lock = new ReentrantLock();

  /**
   * The executor service that runs the cacheCleaner.
   */
  private final ScheduledThreadPoolExecutor cleanerExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder().
          setDaemon(true).setNameFormat("ShortCircuitCache_Cleaner").build());

  /**
   * The executor service that runs the cacheCleaner.
   */
  private final ScheduledThreadPoolExecutor releaserExecutor
      = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder().
          setDaemon(true).setNameFormat("ShortCircuitCache_SlotReleaser").
          build());

  /**
   * A map containing all ShortCircuitReplicaInfo objects, organized by Key.
   * ShortCircuitReplicaInfo objects may contain a replica, or an InvalidToken
   * exception.
   */
  private final HashMap<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>>  replicaInfoMap = new HashMap<ExtendedBlockId,
          Waitable<ShortCircuitReplicaInfo>>();

  /**
   * The CacheCleaner.  We don't create this and schedule it until it becomes
   * necessary.
   */
  private CacheCleaner cacheCleaner;

  /**
   * Tree of evictable elements.
   *
   * Maps (unique) insertion time in nanoseconds to the element.
   */
  private final TreeMap<Long, ShortCircuitReplica> evictable =
      new TreeMap<Long, ShortCircuitReplica>();

  /**
   * Maximum total size of the cache, including both mmapped and
   * no$-mmapped elements.
   */
  private final int maxTotalSize;

  /**
   * Non-mmaped elements older than this will be closed.
   */
  private long maxNonMmappedEvictableLifespanMs;

  /**
   * Tree of mmaped evictable elements.
   *
   * Maps (unique) insertion time in nanoseconds to the element.
   */
  private final TreeMap<Long, ShortCircuitReplica> evictableMmapped =
      new TreeMap<Long, ShortCircuitReplica>();

  /**
   * Maximum number of mmaped evictable elements.
   */
  private int maxEvictableMmapedSize;

  /**
   * Mmaped elements older than this will be closed.
   */
  private final long maxEvictableMmapedLifespanMs;

  /**
   * The minimum number of milliseconds we'll wait after an unsuccessful
   * mmap attempt before trying again.
   */
  private final long mmapRetryTimeoutMs;

  /**
   * How long we will keep replicas in the cache before declaring them
   * to be stale.
   */
  private final long staleThresholdMs;

  /**
   * True if the ShortCircuitCache is closed.
   */
  private boolean closed = false;

  /**
   * Number of existing mmaps associated with this cache.
   */
  private int outstandingMmapCount = 0;

  /**
   * Manages short-circuit shared memory segments for the client.
   */
  private final DfsClientShmManager shmManager;

  /**
   * Create a {@link ShortCircuitCache} object from a {@link Configuration}
   */
  public static ShortCircuitCache fromConf(Configuration conf) {
    return new ShortCircuitCache(
        conf.getInt(DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_STREAMS_CACHE_SIZE_KEY,
            DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_STREAMS_CACHE_SIZE_DEFAULT),
        conf.getLong(DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_STREAMS_CACHE_EXPIRY_MS_KEY,
            DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_STREAMS_CACHE_EXPIRY_MS_DEFAULT),
        conf.getInt(DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_SIZE,
            DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_SIZE_DEFAULT),
        conf.getLong(DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_TIMEOUT_MS,
            DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_TIMEOUT_MS_DEFAULT),
        conf.getLong(DFSConfigKeys.DFS_CLIENT_MMAP_RETRY_TIMEOUT_MS,
            DFSConfigKeys.DFS_CLIENT_MMAP_RETRY_TIMEOUT_MS_DEFAULT),
        conf.getLong(DFSConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_REPLICA_STALE_THRESHOLD_MS,
            DFSConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_REPLICA_STALE_THRESHOLD_MS_DEFAULT),
        conf.getInt(DFSConfigKeys.DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS,
            DFSConfigKeys.DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT));
  }

  public ShortCircuitCache(int maxTotalSize, long maxNonMmappedEvictableLifespanMs,
      int maxEvictableMmapedSize, long maxEvictableMmapedLifespanMs,
      long mmapRetryTimeoutMs, long staleThresholdMs, int shmInterruptCheckMs) {
    Preconditions.checkArgument(maxTotalSize >= 0);
    this.maxTotalSize = maxTotalSize;
    Preconditions.checkArgument(maxNonMmappedEvictableLifespanMs >= 0);
    this.maxNonMmappedEvictableLifespanMs = maxNonMmappedEvictableLifespanMs;
    Preconditions.checkArgument(maxEvictableMmapedSize >= 0);
    this.maxEvictableMmapedSize = maxEvictableMmapedSize;
    Preconditions.checkArgument(maxEvictableMmapedLifespanMs >= 0);
    this.maxEvictableMmapedLifespanMs = maxEvictableMmapedLifespanMs;
    this.mmapRetryTimeoutMs = mmapRetryTimeoutMs;
    this.staleThresholdMs = staleThresholdMs;
    DfsClientShmManager shmManager = null;
    if ((shmInterruptCheckMs > 0) &&
        (DomainSocketWatcher.getLoadingFailureReason() == null)) {
      try {
        shmManager = new DfsClientShmManager(shmInterruptCheckMs);
      } catch (IOException e) {
        LOG.error("failed to create ShortCircuitShmManager", e);
      }
    }
    this.shmManager = shmManager;
  }

  public long getStaleThresholdMs() {
    return staleThresholdMs;
  }

  /**
   * Increment the reference count of a replica, and remove it from any free
   * list it may be in.
   *
   * You must hold the cache lock while calling this function.
   *
   * @param replica      The replica we're removing.
   */
  private void ref(ShortCircuitReplica replica) {
    lock.lock();
    try {
      Preconditions.checkArgument(replica.refCount > 0,
          "can't ref " + replica + " because its refCount reached " +
          replica.refCount);
      Long evictableTimeNs = replica.getEvictableTimeNs();
      replica.refCount++;
      if (evictableTimeNs != null) {
        String removedFrom = removeEvictable(replica);
        if (LOG.isTraceEnabled()) {
          LOG.trace(this + ": " + removedFrom +
              " no longer contains " + replica + ".  refCount " +
              (replica.refCount - 1) + " -> " + replica.refCount +
              StringUtils.getStackTrace(Thread.currentThread()));

        }
      } else if (LOG.isTraceEnabled()) {
        LOG.trace(this + ": replica  refCount " +
            (replica.refCount - 1) + " -> " + replica.refCount +
            StringUtils.getStackTrace(Thread.currentThread()));
      }
    } finally {
      lock.unlock();
    }
  }

  /**
   * unref()操作只能在ShortCircuitReplica.unref()以及purge()方法中被调用，也就是在使用类不再引用ShortCircuitReplica对象以及缓存中不存在ShortCircuitReplica对象时才会被调用。
   * 如果当前副本已经过期或者不可用，unref()方法会直接将当前副本从缓存中删除。
   * 如果当前副本没有被引用，也就是没有类使用这个副本进行读取操作，同时副本已经从缓存中删除了，这种情况就需要将副本关闭。关闭副本操作会释放副本对应的Slot对象，并且释放副本数据块文件的内存映射。
   * 如果副本没有被引用，但在缓存中还存在，则将副本移入evictable队列，向evictable队列中添加了元素后，还需要调用trimEvictionMaps()方法将超过缓存队列大小的副本移出缓存。
   */
  void unref(ShortCircuitReplica replica) {
    lock.lock();
    try {
      // 如果当前副本已经过期或者不可用，则直接将当前副本从缓存中删除
      if (!replica.purged) {
        String purgeReason = null;
        if (!replica.getDataStream().getChannel().isOpen()) {
          purgeReason = "purging replica because its data channel is closed.";
        } else if (!replica.getMetaStream().getChannel().isOpen()) {
          purgeReason = "purging replica because its meta channel is closed.";
        } else if (replica.isStale()) {
          purgeReason = "purging replica because it is stale.";
        }
        if (purgeReason != null) {
          LOG.debug(this + ": " + purgeReason);
          purge(replica);
        }
      }
      String addedString = "";
      boolean shouldTrimEvictionMaps = false;
      // 更新 refCount
      int newRefCount = --replica.refCount;
      if (newRefCount == 0) {
        // 副本没有被引用，也就是没有类使用这个副本进行读取操作，同时副本也从缓存中删除了，这种情况就将副本关闭，释放slot,释放副本数据块文件的内存映射
        Preconditions.checkArgument(replica.purged, "Replica " + replica + " reached a refCount of 0 without " + "being purged");
        replica.close();
      } else if (newRefCount == 1) {
        // 副本没有被引用，也就是没有类使用这个副本进行读取操作，但在缓存中还存在，则将副本移入 evictable 队列
        Preconditions.checkState(null == replica.getEvictableTimeNs(), "Replica " + replica + " had a refCount higher than 1, " + replica.getEvictableTimeNs() + ")");
        if (!replica.purged) {
          // 加入对应的 evictable 队列
          if (replica.hasMmap()) {
            insertEvictable(System.nanoTime(), replica, evictableMmapped);
            addedString = "added to evictableMmapped, ";
          } else {
            insertEvictable(System.nanoTime(), replica, evictable);
            addedString = "added to evictable, ";
          }
          shouldTrimEvictionMaps = true;
        }
      } else {
        Preconditions.checkArgument(replica.refCount >= 0, "replica's refCount went negative (refCount = " + replica.refCount + " for " + replica + ")");
      }
      if (LOG.isTraceEnabled()) {
        LOG.trace(this + ": unref replica " + replica + ": " + addedString + " refCount " + (newRefCount + 1) + " -> " + newRefCount + StringUtils.getStackTrace(Thread.currentThread()));
      }
      // 调整 evictable 队列的大小，大于缓存队列大小的副本直接删除
      if (shouldTrimEvictionMaps) {
        trimEvictionMaps();
      }
    } finally {
      lock.unlock();
    }
  }

  /**
   * Demote old evictable mmaps into the regular eviction map.
   *
   * You must hold the cache lock while calling this function.
   *
   * @param now   Current time in monotonic milliseconds.
   * @return      Number of replicas demoted.
   */
  private int demoteOldEvictableMmaped(long now) {
    int numDemoted = 0;
    boolean needMoreSpace = false;
    Long evictionTimeNs = Long.valueOf(0);

    while (true) {
      Entry<Long, ShortCircuitReplica> entry = 
          evictableMmapped.ceilingEntry(evictionTimeNs);
      if (entry == null) break;
      evictionTimeNs = entry.getKey();
      long evictionTimeMs = 
          TimeUnit.MILLISECONDS.convert(evictionTimeNs, TimeUnit.NANOSECONDS);
      if (evictionTimeMs + maxEvictableMmapedLifespanMs >= now) {
        if (evictableMmapped.size() < maxEvictableMmapedSize) {
          break;
        }
        needMoreSpace = true;
      }
      ShortCircuitReplica replica = entry.getValue();
      if (LOG.isTraceEnabled()) {
        String rationale = needMoreSpace ? "because we need more space" : 
            "because it's too old";
        LOG.trace("demoteOldEvictable: demoting " + replica + ": " +
            rationale + ": " +
            StringUtils.getStackTrace(Thread.currentThread()));
      }
      removeEvictable(replica, evictableMmapped);
      munmap(replica);
      insertEvictable(evictionTimeNs, replica, evictable);
      numDemoted++;
    }
    return numDemoted;
  }

  /**
   * Trim the eviction lists.
   */
  private void trimEvictionMaps() {
    long now = Time.monotonicNow();
    demoteOldEvictableMmaped(now);

    while (true) {
      long evictableSize = evictable.size();
      long evictableMmappedSize = evictableMmapped.size();
      if (evictableSize + evictableMmappedSize <= maxTotalSize) {
        return;
      }
      ShortCircuitReplica replica;
      if (evictableSize == 0) {
       replica = evictableMmapped.firstEntry().getValue();
      } else {
       replica = evictable.firstEntry().getValue();
      }
      if (LOG.isTraceEnabled()) {
        LOG.trace(this + ": trimEvictionMaps is purging " + replica +
          StringUtils.getStackTrace(Thread.currentThread()));
      }
      purge(replica);
    }
  }

  /**
   * Munmap a replica, updating outstandingMmapCount.
   *
   * @param replica  The replica to munmap.
   */
  private void munmap(ShortCircuitReplica replica) {
    replica.munmap();
    outstandingMmapCount--;
  }

  /**
   * Remove a replica from an evictable map.
   *
   * @param replica   The replica to remove.
   * @return          The map it was removed from.
   */
  private String removeEvictable(ShortCircuitReplica replica) {
    if (replica.hasMmap()) {
      removeEvictable(replica, evictableMmapped);
      return "evictableMmapped";
    } else {
      removeEvictable(replica, evictable);
      return "evictable";
    }
  }

  /**
   * Remove a replica from an evictable map.
   *
   * @param replica   The replica to remove.
   * @param map       The map to remove it from.
   */
  private void removeEvictable(ShortCircuitReplica replica,
      TreeMap<Long, ShortCircuitReplica> map) {
    Long evictableTimeNs = replica.getEvictableTimeNs();
    Preconditions.checkNotNull(evictableTimeNs);
    ShortCircuitReplica removed = map.remove(evictableTimeNs);
    Preconditions.checkState(removed == replica,
        "failed to make " + replica + " unevictable");
    replica.setEvictableTimeNs(null);
  }

  /**
   * Insert a replica into an evictable map.
   *
   * If an element already exists with this eviction time, we add a nanosecond
   * to it until we find an unused key.
   *
   * @param evictionTimeNs   The eviction time in absolute nanoseconds.
   * @param replica          The replica to insert.
   * @param map              The map to insert it into.
   */
  private void insertEvictable(Long evictionTimeNs,
      ShortCircuitReplica replica, TreeMap<Long, ShortCircuitReplica> map) {
    while (map.containsKey(evictionTimeNs)) {
      evictionTimeNs++;
    }
    Preconditions.checkState(null == replica.getEvictableTimeNs());
    replica.setEvictableTimeNs(evictionTimeNs);
    map.put(evictionTimeNs, replica);
  }

  /**
   *  fetch() 和 create() 方法都用于在缓存中添加ShortCircuitReplica对象，而purge()方法是从缓存中删除ShortCircuitReplica。
   *
   *  要注意purge()方法只是将副本从缓存中移除，并没有关闭ShortCircuitReplica对象，也就是没有释放共享内存中的槽位。
   *  Purge()方法首先会设置副本的状态为purged，然后将ShortCircuitReplica对象从replicaInfoMap以及evictable队列中删除。最后由于已经从缓存中删除了ShortCircuitReplica对象，
   *  所以调用unref()方法降低副本的引用计数。
   */
  private void purge(ShortCircuitReplica replica) {
    boolean removedFromInfoMap = false;
    String evictionMapName = null;
    Preconditions.checkArgument(!replica.purged);
    // replica已经删除
    replica.purged = true;
    Waitable<ShortCircuitReplicaInfo> val = replicaInfoMap.get(replica.key);
    // 从 replicaInfoMap中删除数据
    if (val != null) {
      ShortCircuitReplicaInfo info = val.getVal();
      if ((info != null) && (info.getReplica() == replica)) {
        replicaInfoMap.remove(replica.key);
        removedFromInfoMap = true;
      }
    }
    // 从evictable队列中删除副本数据
    Long evictableTimeNs = replica.getEvictableTimeNs();
    if (evictableTimeNs != null) {
      evictionMapName = removeEvictable(replica);
    }
    if (LOG.isTraceEnabled()) {
      StringBuilder builder = new StringBuilder();
      builder.append(this).append(": ").append(": purged "). append(replica).append(" from the cache.");
      if (removedFromInfoMap) {
        builder.append("  Removed from the replicaInfoMap.");
      }
      if (evictionMapName != null) {
        builder.append("  Removed from ").append(evictionMapName);
      }
      LOG.trace(builder.toString());
    }
    // 由于在replicag构造时就考虑了缓存的引用，所以从缓存中删除时，要 unref() 这个replica
    unref(replica); // TODO 核心方法
  }

  /**
   * fetchOrCreate()方法用于从ShortCircuitCache缓存中获取一个ShortCircuitReplicaInfo对象（封装了ShortCircuitReplica对象），或者在缓存没有保存这个对象时创建一个新的ShortCircuitReplicaInfo对象。
   *
   * 需要特别注意的是，可能同时有多个线程并发地调用这个方法来获取同一个ShortCircuit ReplicaInfo对象，
   * fetchOrCreate()会允许第一个申请创建ShortCircuitReplicaInfo对象的线程调用create()方法执行创建操作，
   * 同时fetchOrCreate()方法会在缓存ShortCircuitReplicaInfo对象的ShortCircuitCache.replicaInfoMap字段中放入一个Waitable对象。
   * 当其他的线程调用fetchOrCreate()方法申请获取同一个ShortCircuitReplicaInfo对象时，会得到这个Waitable对象并在这个Waitable对象上等待。
   * 当创建线程成功地创建ShortCircuitReplicaInfo对象后，会唤醒该线程并通过Waitable对象将ShortCircuitReplicaInfo传递给该线程。
   * 通过这种同步机制，多个线程之间可以并发地访问缓存，值得我们学习。
   */
  public ShortCircuitReplicaInfo fetchOrCreate(ExtendedBlockId key, ShortCircuitReplicaCreator creator) {
    Waitable<ShortCircuitReplicaInfo> newWaitable = null;
    lock.lock();
    try {
      ShortCircuitReplicaInfo info = null;
      do {
        if (closed) {
          return null;
        }
        // 这里从 replicaInfoMap 中获取 Waitable对象
        Waitable<ShortCircuitReplicaInfo> waitable = replicaInfoMap.get(key);
        if (waitable != null) { // 已经有线程构造 ShortCircuitReplicaInfo 对象的情况
          try {
            // 调用 fetch() 方法等待获取 ShortCircuitReplicaInfo 对象
            info = fetch(key, waitable); // TODO 核心代码
          } catch (RetriableException e) {
            continue;
          }
        }
      } while (false);
      if (info != null) {
        return info;
      }
      // 第一个尝试创建 ShortCircuitReplicaInfo 对象的西安测绘给你，先构造 Waitable并放入 plicaInfoMap字段中
      newWaitable = new Waitable<ShortCircuitReplicaInfo>(lock.newCondition());
      replicaInfoMap.put(key, newWaitable);
    } finally {
      lock.unlock();
    }
    // 调用create() 方法 创建ShortCircuitReplicaInfo 对象 并返回
    return create(key, creator, newWaitable); // TODO 核心代码
  }

  /**
   * fetch()方法会在Waitable对象上调用await()方法，等待另一个线程完成ShortCircuitReplica对象的创建操作。
   *
   * 这里需要注意的是，对于purged 以及 stale状态的ShortCircuitReplica对象，fetch()方法会抛出RetriableException异常。
   * 同时由于fetch()方法的调用类会获取ShortCircuitReplica对象的引用，所以还需要调用ref()操作增加副本的引用次数。
   */
  private ShortCircuitReplicaInfo fetch(ExtendedBlockId key, Waitable<ShortCircuitReplicaInfo> waitable) throws RetriableException {
    // Another thread is already in the process of loading this ShortCircuitReplica.  So we simply wait for it to complete.
    ShortCircuitReplicaInfo info;
    try {
      // 在Waitable对象上等待，等待另一个线程构造 ShortCircuitReplicaInfo 对象
      info = waitable.await();
    } catch (InterruptedException e) {
      LOG.info(this + ": interrupted while waiting for " + key);
      Thread.currentThread().interrupt();
      throw new RetriableException("interrupted");
    }

    if (info.getInvalidTokenException() != null) {
      LOG.warn(this + ": could not get " + key + " due to InvalidToken " + "exception.", info.getInvalidTokenException());
      return info;
    }

    ShortCircuitReplica replica = info.getReplica();
    if (replica == null) {
      LOG.warn(this + ": failed to get " + key);
      return info;
    }
    if (replica.purged) {
      // 如果当前取得的副本已经从缓存中清除了，则抛出异常。在 fetchAndCreate() 方法中调用 create() 创建一个新的副本
      throw new RetriableException("Ignoring purged replica " + replica + ".  Retrying.");
    }
    // 如果当前副本已经过期了，则从缓存中清除该副本，抛出异常，在 fetchAndCreate() 方法中 调用 create() 创建一个新的副本
    if (replica.isStale()) {
      LOG.info(this + ": got stale replica " + replica + ".  Removing " + "this replica from the replicaInfoMap and retrying.");
      purge(replica);
      throw new RetriableException("ignoring stale replica " + replica);
    }
    // 在副本上添加引用计数，因为是获取一个已经创建的副本，所以需要添加新的引用计数
    ref(replica);
    return info;
  }

  /**
   * 它首先会调用BlockReaderFacotry.createShortCircuitReplicaInfo()方法创建ShortCircuitReplicaInfo对象，
   * 创建成功之后会将新创建的ShortCircuitReplicaInfo 添加到Waitable对象中，并且唤醒所有在Waitable上等待的线程。
   */
  private ShortCircuitReplicaInfo create(ExtendedBlockId key, ShortCircuitReplicaCreator creator, Waitable<ShortCircuitReplicaInfo> newWaitable) {
    // Handle loading a new replica.
    ShortCircuitReplicaInfo info = null;
    try {
      // 这里调用了 BlockReaderFacotry.createShortCircuitReplicaInfo() 方法创建 ShortCircuitReplicaInfo 对象
      info = creator.createShortCircuitReplicaInfo(); // TODO
    } catch (RuntimeException e) {
      LOG.warn(this + ": failed to load " + key, e);
    }
    // 否则，调用构造方法构造这个 ShortCircuitReplicaInfo 对象
    if (info == null) {
      info = new ShortCircuitReplicaInfo();
    }
    lock.lock();
    try {
      if (info.getReplica() != null) {
        // 创建成功，确保启动了 CacheCleaner
        startCacheCleanerThreadIfNeeded();
      } else {
        // 创建失败，从 replicaInfoMap中删除Waitable
        Waitable<ShortCircuitReplicaInfo> waitableInMap = replicaInfoMap.get(key);
        if (waitableInMap == newWaitable) {
          replicaInfoMap.remove(key);
        }
        if (info.getInvalidTokenException() != null) {
          LOG.warn(this + ": could not load " + key + " due to InvalidToken " + "exception.", info.getInvalidTokenException());
        } else {
          LOG.warn(this + ": failed to load " + key);
        }
      }
      // 将 ShortCircuitReplicaInfo 添加到 Waitable中，并且唤醒所有在Waitable上等待的线程
      newWaitable.provide(info);
    } finally {
      lock.unlock();
    }
    return info;
  }

  private void startCacheCleanerThreadIfNeeded() {
    if (cacheCleaner == null) {
      cacheCleaner = new CacheCleaner();
      long rateMs = cacheCleaner.getRateInMs();
      ScheduledFuture<?> future =
          cleanerExecutor.scheduleAtFixedRate(cacheCleaner, rateMs, rateMs,
              TimeUnit.MILLISECONDS);
      cacheCleaner.setFuture(future);
      if (LOG.isDebugEnabled()) {
        LOG.debug(this + ": starting cache cleaner thread which will run " +
          "every " + rateMs + " ms");
      }
    }
  }

  ClientMmap getOrCreateClientMmap(ShortCircuitReplica replica, boolean anchored) {
    Condition newCond;
    lock.lock();
    try {
      while (replica.mmapData != null) {
        if (replica.mmapData instanceof MappedByteBuffer) {
          ref(replica);
          MappedByteBuffer mmap = (MappedByteBuffer)replica.mmapData;
          return new ClientMmap(replica, mmap, anchored);
        } else if (replica.mmapData instanceof Long) {
          long lastAttemptTimeMs = (Long)replica.mmapData;
          long delta = Time.monotonicNow() - lastAttemptTimeMs;
          if (delta < mmapRetryTimeoutMs) {
            return null;
          }
        } else if (replica.mmapData instanceof Condition) {
          Condition cond = (Condition)replica.mmapData;
          cond.awaitUninterruptibly();
        } else {
          Preconditions.checkState(false, "invalid mmapData type " + replica.mmapData.getClass().getName());
        }
      }
      newCond = lock.newCondition();
      replica.mmapData = newCond;
    } finally {
      lock.unlock();
    }
    MappedByteBuffer map = replica.loadMmapInternal(); // TODO 核心代码
    lock.lock();
    try {
      if (map == null) {
        replica.mmapData = Long.valueOf(Time.monotonicNow());
        newCond.signalAll();
        return null;
      } else {
        outstandingMmapCount++;
        replica.mmapData = map;
        ref(replica);
        newCond.signalAll();
        return new ClientMmap(replica, map, anchored);
      }
    } finally {
      lock.unlock();
    }
  }

  /**
   * Close the cache and free all associated resources.
   */
  @Override
  public void close() {
    try {
      lock.lock();
      if (closed) return;
      closed = true;
      LOG.info(this + ": closing");
      maxNonMmappedEvictableLifespanMs = 0;
      maxEvictableMmapedSize = 0;
      // Close and join cacheCleaner thread.
      IOUtils.cleanup(LOG, cacheCleaner);
      // Purge all replicas.
      while (true) {
        Entry<Long, ShortCircuitReplica> entry = evictable.firstEntry();
        if (entry == null) break;
        purge(entry.getValue());
      }
      while (true) {
        Entry<Long, ShortCircuitReplica> entry = evictableMmapped.firstEntry();
        if (entry == null) break;
        purge(entry.getValue());
      }
    } finally {
      lock.unlock();
    }
    IOUtils.cleanup(LOG, shmManager);
  }

  @VisibleForTesting // ONLY for testing
  public interface CacheVisitor {
    void visit(int numOutstandingMmaps,
        Map<ExtendedBlockId, ShortCircuitReplica> replicas,
        Map<ExtendedBlockId, InvalidToken> failedLoads,
        Map<Long, ShortCircuitReplica> evictable,
        Map<Long, ShortCircuitReplica> evictableMmapped);
  }

  @VisibleForTesting // ONLY for testing
  public void accept(CacheVisitor visitor) {
    lock.lock();
    try {
      Map<ExtendedBlockId, ShortCircuitReplica> replicas =
          new HashMap<ExtendedBlockId, ShortCircuitReplica>();
      Map<ExtendedBlockId, InvalidToken> failedLoads =
          new HashMap<ExtendedBlockId, InvalidToken>();
      for (Entry<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>> entry :
            replicaInfoMap.entrySet()) {
        Waitable<ShortCircuitReplicaInfo> waitable = entry.getValue();
        if (waitable.hasVal()) {
          if (waitable.getVal().getReplica() != null) {
            replicas.put(entry.getKey(), waitable.getVal().getReplica());
          } else {
            // The exception may be null here, indicating a failed load that
            // isn't the result of an invalid block token.
            failedLoads.put(entry.getKey(),
                waitable.getVal().getInvalidTokenException());
          }
        }
      }
      if (LOG.isDebugEnabled()) {
        StringBuilder builder = new StringBuilder();
        builder.append("visiting ").append(visitor.getClass().getName()).
            append("with outstandingMmapCount=").append(outstandingMmapCount).
            append(", replicas=");
        String prefix = "";
        for (Entry<ExtendedBlockId, ShortCircuitReplica> entry : replicas.entrySet()) {
          builder.append(prefix).append(entry.getValue());
          prefix = ",";
        }
        prefix = "";
        builder.append(", failedLoads=");
        for (Entry<ExtendedBlockId, InvalidToken> entry : failedLoads.entrySet()) {
          builder.append(prefix).append(entry.getValue());
          prefix = ",";
        }
        prefix = "";
        builder.append(", evictable=");
        for (Entry<Long, ShortCircuitReplica> entry : evictable.entrySet()) {
          builder.append(prefix).append(entry.getKey()).
              append(":").append(entry.getValue());
          prefix = ",";
        }
        prefix = "";
        builder.append(", evictableMmapped=");
        for (Entry<Long, ShortCircuitReplica> entry : evictableMmapped.entrySet()) {
          builder.append(prefix).append(entry.getKey()).
              append(":").append(entry.getValue());
          prefix = ",";
        }
        LOG.debug(builder.toString());
      }
      visitor.visit(outstandingMmapCount, replicas, failedLoads,
            evictable, evictableMmapped);
    } finally {
      lock.unlock();
    }
  }

  @Override
  public String toString() {
    return "ShortCircuitCache(0x" +
        Integer.toHexString(System.identityHashCode(this)) + ")";
  }

  /**
   * Allocate a new shared memory slot.
   *
   * @param datanode       The datanode to allocate a shm slot with.
   * @param peer           A peer connected to the datanode.
   * @param usedPeer       Will be set to true if we use up the provided peer.
   * @param blockId        The block id and block pool id of the block we're 
   *                         allocating this slot for.
   * @param clientName     The name of the DFSClient allocating the shared
   *                         memory.
   * @return               Null if short-circuit shared memory is disabled;
   *                         a short-circuit memory slot otherwise.
   * @throws IOException   An exception if there was an error talking to 
   *                         the datanode.
   */
  public Slot allocShmSlot(DatanodeInfo datanode,
        DomainPeer peer, MutableBoolean usedPeer,
        ExtendedBlockId blockId, String clientName) throws IOException {
    if (shmManager != null) {
      return shmManager.allocSlot(datanode, peer, usedPeer, blockId, clientName); // TODO
    } else {
      return null;
    }
  }

  /**
   * Free a slot immediately.
   *
   * ONLY use this if the DataNode is not yet aware of the slot.
   * 
   * @param slot           The slot to free.
   */
  public void freeSlot(Slot slot) {
    Preconditions.checkState(shmManager != null);
    slot.makeInvalid();
    shmManager.freeSlot(slot);
  }
  
  /**
   * 它首先调用DataTransferProtocol.releaseShortCircuitFds()方法通知Datanode释放Datanode侧共享内存中的槽位，
   * 然后调用shmManager.freeSlot()方法释放Client侧共享内存中的槽位。
   * 如果在RPC过程中失败，则直接关闭这个共享内存段，因为Client与Datanode之间的通信已经出现了异常。
   */
  public void scheduleSlotReleaser(Slot slot) {
    Preconditions.checkState(shmManager != null);
    releaserExecutor.execute(new SlotReleaser(slot));
  }

  @VisibleForTesting
  public DfsClientShmManager getDfsClientShmManager() {
    return shmManager;
  }
}
